package cn.krguang;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.json.JSONException;
import org.json.JSONObject;

import java.util.ArrayList;
import java.util.HashMap;


public class MyMqttClient  {

    private String topic = "device/#";
    private String broker = "tcp://www.krguang.cn:1883";
    private String userName = "krg";
    private String password = "Mqtt17612168015";
    private String clientId = "86756234677333343";
    private MqttClient mqttClient;
    private static MqttConnectOptions mqttConnectOptions;
    private boolean ConnectFlage = true;

    private OnServerConnectedCallback ConnectedCallback;//连接到服务器
    private OnServerDisConnectedCallback DisConnectedCallback;//与服务器断开连接
    private OnServerReadStringCallback ReadStringCallback;//接收信息回调(字符串)

    private MyMqttClient(){
    }
    private static MyMqttClient instance = new MyMqttClient();
    public static MyMqttClient getInstance(){
        return instance;
    }

    private void InitMqttOptions(){
        mqttConnectOptions = new MqttConnectOptions();//MQTT的连接设置
        //设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        mqttConnectOptions.setCleanSession(true);
        mqttConnectOptions.setUserName(userName);//设置连接的用户名
        mqttConnectOptions.setPassword(password.toCharArray());//设置连接的密码
        mqttConnectOptions.setConnectionTimeout(3);// 设置连接超时时间 单位为秒
        // 设置会话心跳时间 单位为秒
        mqttConnectOptions.setKeepAliveInterval(60);
    }

    /*初始化Mqtt连接*/
    private void InitMqttConnect() {
        try {

            mqttClient = new MqttClient(broker,clientId,new MemoryPersistence());
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void messageArrived(String arg0, MqttMessage arg1){
                    if (ReadStringCallback!=null){
                        ReadStringCallback.callback(arg0,arg1);
                    }
                }
                @Override
                public void deliveryComplete(IMqttDeliveryToken arg0) { }
                @Override
                public void connectionLost(Throwable arg0) {
                    if (DisConnectedCallback!=null)
                        DisConnectedCallback.callback(arg0);
                    try {
                        mqttClient.disconnect();
                    }
                    catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                    try {
                        mqttClient.close();
                    } catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                    mqttClient = null;
                    ConnectFlage = true;
                    setConnect();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void setConnect(){
        if (mqttClient == null || !mqttClient.isConnected()) {
            InitMqttConnect();
            InitMqttOptions();
        }else {
            System.out.println("setConnect: mqttClient Is Connected");
        }
        new Thread(new Runnable() {
            @Override
            public void run() {
                while(ConnectFlage) {
                    try {
                        if (mqttClient == null || !mqttClient.isConnected()) {
                            System.out.println("run: Connect ......");
                            mqttClient.connect(mqttConnectOptions);
                            System.out.println("Connect Success??");
                            if (mqttClient.isConnected()) {
                                if (ConnectedCallback != null)
                                    ConnectedCallback.callback();
                                ConnectFlage = false;
                                System.out.println("Connect Success");
                                setSubscribe(topic, 0);
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        if (DisConnectedCallback!=null)
                            DisConnectedCallback.callback(e);
                        try{
                            Thread.sleep(3000);
                        }catch (Exception e1){
                            e1.printStackTrace();
                        }
                    }
                }
            }
        }).start();
    }

    //订阅主题
    public void setSubscribe(final String Topic, final int qos) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try{
                    if (mqttClient!=null){
                        mqttClient.subscribe(Topic,qos);
                    }
                }
                catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }


    public interface OnServerConnectedCallback {
        void callback();
    }
    public interface OnServerDisConnectedCallback {
        void callback(Throwable e);
    }
    public interface OnServerReadStringCallback {
        void callback(String Topic, MqttMessage Msg);
    }

    public void setOnServerReadStringCallback(OnServerReadStringCallback ReadStringCallback) {
        this.ReadStringCallback = ReadStringCallback;
    }

}